package com.fengye.hbase.example;

import java.io.IOException;
import java.util.*;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
import org.apache.hadoop.hbase.util.Bytes;

public class Example {

    private static final String TABLE_NAME = "MY_TABLE_NAME_TOO";
    private static final String CF_DEFAULT = "DEFAULT_COLUMN_FAMILY";

    public static void createOrOverwrite(Admin admin, HTableDescriptor table) throws IOException {
        if (admin.tableExists(table.getTableName())) {
            try {
                admin.enableTable(table.getTableName());
            } catch (TableNotDisabledException e) {
                // e.printStackTrace();
                System.out.println("hehe");
            } finally {
                admin.disableTable(table.getTableName());
                admin.deleteTable(table.getTableName());
            }
        }
        admin.createTable(table);
    }

    public static void createSchemaTables(Configuration config) throws IOException {
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {

            HTableDescriptor table = new HTableDescriptor(TableName.valueOf(TABLE_NAME));
            table.addFamily(new HColumnDescriptor(CF_DEFAULT).setCompressionType(Algorithm.NONE));

            System.out.print("Creating table. ");
            createOrOverwrite(admin, table);
            System.out.println(" Done.");
        }
    }

    public static void modifySchema(Configuration config) throws IOException {
        try (Connection connection = ConnectionFactory.createConnection(config);
             Admin admin = connection.getAdmin()) {

            TableName tableName = TableName.valueOf(TABLE_NAME);
            if (!admin.tableExists(tableName)) {
                System.out.println("Table does not exist.");
                System.exit(-1);
            }

            HTableDescriptor table = admin.getTableDescriptor(tableName);

            // Update existing table
            HColumnDescriptor newColumn = new HColumnDescriptor("NEWCF");
            newColumn.setCompactionCompressionType(Algorithm.GZ);
            newColumn.setMaxVersions(HConstants.ALL_VERSIONS);
            admin.addColumn(tableName, newColumn);

            // Update existing column family
            HColumnDescriptor existingColumn = new HColumnDescriptor(CF_DEFAULT);
            existingColumn.setCompactionCompressionType(Algorithm.GZ);
            existingColumn.setMaxVersions(HConstants.ALL_VERSIONS);
            table.modifyFamily(existingColumn);
            admin.modifyTable(tableName, table);

            // add new column
            HColumnDescriptor newColumnF = new HColumnDescriptor("f");
            newColumnF.setCompactionCompressionType(Algorithm.GZ);
            newColumnF.setMaxVersions(HConstants.ALL_VERSIONS);
            admin.addColumn(tableName, newColumnF);

            HColumnDescriptor[] columnFamilies = table.getColumnFamilies();
            for (HColumnDescriptor columnFamily : columnFamilies) {
                System.out.println("table COLUMN_FAMILY : " + Bytes.toString(columnFamily.getName()));
                // Map<ImmutableBytesWritable, ImmutableBytesWritable> values = columnFamily.getValues();
                // for (ImmutableBytesWritable immutableBytesWritable : values.keySet()) {
                //     System.out.println(Bytes.toString(immutableBytesWritable.get()));
                //     for (Map.Entry<ImmutableBytesWritable, ImmutableBytesWritable> immutableBytesWritableImmutableBytesWritableEntry : values.entrySet()) {
                //         System.out.println(Bytes.toString(immutableBytesWritableImmutableBytesWritableEntry.getKey().get())+":"+
                //         Bytes.toString(immutableBytesWritableImmutableBytesWritableEntry.getValue().get()));
                //     }
                // }
            }

            Table table1 = connection.getTable(tableName);
            Put put = new Put("row1".getBytes());
            put.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "id".getBytes(), "1".getBytes());
            put.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "name".getBytes(), "zhangsan".getBytes());
            put.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "age".getBytes(), "15".getBytes());
            put.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "school".getBytes(), "victory".getBytes());
            put.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "sex".getBytes(), "female".getBytes());

            Put put1 = new Put("row2".getBytes());
            put1.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "id".getBytes(), "2".getBytes());
            put1.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "name".getBytes(), "lisi".getBytes());
            put1.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "age".getBytes(), "16".getBytes());
            put1.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "school".getBytes(), "tuanjie".getBytes());
            put1.addColumn("DEFAULT_COLUMN_FAMILY".getBytes(), "sex".getBytes(), "male".getBytes());


            Put put3 = new Put("row1".getBytes());
            put3.addColumn("f".getBytes(), "id".getBytes(), "1".getBytes());
            put3.addColumn("f".getBytes(), "name".getBytes(), "zhangsan".getBytes());
            put3.addColumn("f".getBytes(), "age".getBytes(), "15".getBytes());
            put3.addColumn("f".getBytes(), "school".getBytes(), "victory".getBytes());
            put3.addColumn("f".getBytes(), "sex".getBytes(), "female".getBytes());


            ArrayList<Put> putArrayList = new ArrayList<>();
            putArrayList.add(put);
            putArrayList.add(put1);
            putArrayList.add(put3);
            table1.put(putArrayList);

            ResultScanner results = table1.getScanner(new Scan());
            Result result;
            System.out.println("遍历表中的数据");
            while (null != (result = results.next())) {
                // System.out.println("row:" + Bytes.toString((result.getRow())));
                NavigableMap<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> map = result.getMap();
                for (Map.Entry<byte[], NavigableMap<byte[], NavigableMap<Long, byte[]>>> mapEntry : map.entrySet()) {
                    byte[] mapEntryKey = mapEntry.getKey();
                    // System.out.println(Bytes.toString(mapEntryKey));
                    Set<Map.Entry<byte[], NavigableMap<Long, byte[]>>> entries = mapEntry.getValue().entrySet();
                    for (Map.Entry<byte[], NavigableMap<Long, byte[]>> entry : entries) {
                        System.out.println(Bytes.toString((result.getRow())) + ":" + Bytes.toString(mapEntryKey) + ":" + Bytes.toString(entry.getKey()) + ":" +
                                Bytes.toString(entry.getValue().firstEntry().getValue()));
                    }
                }
            }
            // Disable an existing table
            admin.disableTable(tableName);

            // Delete an existing column family
            // 本表中只有一列是不能删除的.
            // admin.deleteColumn(tableName, CF_DEFAULT.getBytes("UTF-8"));

            // Delete a table (Need to be disabled first)
            admin.deleteTable(tableName);
        }
    }

    public static void main(String... args) throws IOException {
        Configuration config = HBaseConfiguration.create();

        config.set("hbase.zookeeper.quorum", "hadoop");
        createSchemaTables(config);
        modifySchema(config);

    }
}
